package juc.scenarios.datafetch;

import lombok.Getter;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 并发数据抓取系统演示
 * 
 * 技术点：Semaphore + 线程池
 * 业务场景：从多个数据源并发抓取数据，限制并发数量避免对数据源造成过大压力
 * 
 * 功能特点：
 * 1. 使用Semaphore控制并发访问数量
 * 2. 结合线程池管理抓取任务
 * 3. 实现数据的聚合和结果处理
 * 4. 支持超时控制和异常处理
 * 5. 提供实时监控和统计
 */
public class ConcurrentDataFetchDemo {
    
    // 数据源列表
    private static final List<String> DATA_SOURCES = Arrays.asList(
        "用户服务API", "订单服务API", "商品服务API", "库存服务API", 
        "支付服务API", "物流服务API", "推荐服务API", "营销服务API",
        "客服服务API", "统计服务API", "日志服务API", "配置服务API"
    );
    
    // 控制并发数量
    private static final int MAX_CONCURRENT_FETCHES = 3;
    
    // 信号量控制并发访问
    private final Semaphore semaphore;
    
    // 线程池管理抓取任务
    private final ExecutorService executor;
    
    // 统计信息
    private final AtomicInteger successCount = new AtomicInteger(0);
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicLong totalFetchTime = new AtomicLong(0);
    
    // 结果存储
    private final Map<String, FetchResult> fetchResults = new ConcurrentHashMap<>();
    
    public ConcurrentDataFetchDemo() {
        this.semaphore = new Semaphore(MAX_CONCURRENT_FETCHES);
        this.executor = new ThreadPoolExecutor(
            MAX_CONCURRENT_FETCHES, 
            MAX_CONCURRENT_FETCHES,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(),
            new ThreadFactory() {
                private final AtomicInteger threadNumber = new AtomicInteger(1);
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "DataFetch-Worker-" + threadNumber.getAndIncrement());
                    t.setDaemon(false);
                    return t;
                }
            },
            new ThreadPoolExecutor.AbortPolicy()
        );
    }
    
    /**
     * 数据抓取结果
     */
    @Getter
    public static class FetchResult {
        // Getters
        private final String dataSource;
        private final boolean success;
        private final String data;
        private final long fetchTime;
        private final String error;
        private final long timestamp;
        
        public FetchResult(String dataSource, boolean success, String data, long fetchTime, String error) {
            this.dataSource = dataSource;
            this.success = success;
            this.data = data;
            this.fetchTime = fetchTime;
            this.error = error;
            this.timestamp = System.currentTimeMillis();
        }

        @Override
        public String toString() {
            return String.format("%s: %s (耗时: %dms)%s",
                dataSource,
                success ? "成功" : "失败",
                fetchTime,
                success ? "" : ", 错误: " + error
            );
        }
    }
    
    /**
     * 模拟数据抓取任务
     */
    private class DataFetchTask implements Callable<FetchResult> {
        private final String dataSource;
        
        public DataFetchTask(String dataSource) {
            this.dataSource = dataSource;
        }
        
        @Override
        public FetchResult call() {
            String threadName = Thread.currentThread().getName();
            System.out.printf("[%s] 开始抓取: %s (可用许可证: %d)%n", 
                threadName, dataSource, semaphore.availablePermits());
            
            long startTime = System.currentTimeMillis();
            
            try {
                // 模拟数据抓取过程
                int fetchTime = simulateDataFetch();
                
                // 模拟随机失败
                if (Math.random() < 0.15) { // 15%失败率
                    throw new RuntimeException("网络超时");
                }
                
                // 生成模拟数据
                String data = generateMockData(dataSource);
                
                long endTime = System.currentTimeMillis();
                long duration = endTime - startTime;
                
                FetchResult result = new FetchResult(dataSource, true, data, duration, null);
                
                // 更新统计
                successCount.incrementAndGet();
                totalFetchTime.addAndGet(duration);
                
                System.out.printf("[%s] 完成抓取: %s (耗时: %dms)%n", 
                    threadName, dataSource, duration);
                
                return result;
                
            } catch (Exception e) {
                long endTime = System.currentTimeMillis();
                long duration = endTime - startTime;
                
                FetchResult result = new FetchResult(dataSource, false, null, duration, e.getMessage());
                
                // 更新统计
                failureCount.incrementAndGet();
                
                System.err.printf("[%s] 抓取失败: %s (错误: %s)%n", 
                    threadName, dataSource, e.getMessage());
                
                return result;
                
            } finally {
                // 释放许可证
                semaphore.release();
                System.out.printf("[%s] 释放许可证: %s (可用许可证: %d)%n", 
                    threadName, dataSource, semaphore.availablePermits());
            }
        }
        
        /**
         * 模拟数据抓取时间
         */
        private int simulateDataFetch() throws InterruptedException {
            // 模拟不同数据源的抓取时间（300-1500ms）
            int baseTime = 300 + (dataSource.hashCode() % 1200);
            int fetchTime = baseTime + (int)(Math.random() * 200);
            
            Thread.sleep(fetchTime);
            return fetchTime;
        }
        
        /**
         * 生成模拟数据
         */
        private String generateMockData(String dataSource) {
            Random random = new Random();
            StringBuilder data = new StringBuilder();
            
            if (dataSource.contains("用户")) {
                data.append("用户数据: ")
                    .append("用户ID=").append(random.nextInt(10000))
                    .append(", 用户名=user").append(random.nextInt(1000))
                    .append(", 状态=活跃");
            } else if (dataSource.contains("订单")) {
                data.append("订单数据: ")
                    .append("订单ID=ORD-").append(random.nextInt(100000))
                    .append(", 金额=").append(String.format("%.2f", random.nextDouble() * 1000))
                    .append(", 状态=已完成");
            } else if (dataSource.contains("商品")) {
                data.append("商品数据: ")
                    .append("商品ID=").append(random.nextInt(10000))
                    .append(", 名称=商品").append(random.nextInt(1000))
                    .append(", 库存=").append(random.nextInt(1000));
            } else {
                data.append(dataSource).append("数据: ")
                    .append("ID=").append(random.nextInt(10000))
                    .append(", 状态=正常")
                    .append(", 时间=").append(new Date());
            }
            
            return data.toString();
        }
    }
    
    /**
     * 执行并发数据抓取
     */
    public Map<String, FetchResult> fetchDataConcurrently() {
        System.out.println("\n🚀 并发数据抓取系统启动");
        System.out.println("数据源数量: " + DATA_SOURCES.size());
        System.out.println("最大并发数: " + MAX_CONCURRENT_FETCHES);
        System.out.println("=".repeat(60));
        
        long startTime = System.currentTimeMillis();
        List<Future<FetchResult>> futures = new ArrayList<>();
        
        // 提交所有抓取任务
        for (String dataSource : DATA_SOURCES) {
            try {
                // 获取许可证（可能阻塞）
                semaphore.acquire();
                System.out.println("获取许可证: " + dataSource + " (可用许可证: " + semaphore.availablePermits() + ")");
                
                // 提交任务到线程池
                Future<FetchResult> future = executor.submit(new DataFetchTask(dataSource));
                futures.add(future);
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("线程被中断: " + e.getMessage());
                semaphore.release(); // 释放许可证
            }
        }
        
        // 等待所有任务完成
        System.out.println("\n⏳ 等待所有数据抓取任务完成...");
        
        for (Future<FetchResult> future : futures) {
            try {
                FetchResult result = future.get(5, TimeUnit.SECONDS); // 5秒超时
                fetchResults.put(result.getDataSource(), result);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("任务被中断");
            } catch (ExecutionException e) {
                System.err.println("任务执行异常: " + e.getCause().getMessage());
            } catch (TimeoutException e) {
                System.err.println("任务超时");
                future.cancel(true);
            }
        }
        
        long endTime = System.currentTimeMillis();
        long totalTime = endTime - startTime;
        
        // 输出统计信息
        printStatistics(totalTime);
        
        return fetchResults;
    }
    
    /**
     * 输出统计信息
     */
    private void printStatistics(long totalTime) {
        System.out.println("\n📊 数据抓取统计");
        System.out.println("=".repeat(60));
        
        int total = DATA_SOURCES.size();
        int success = successCount.get();
        int failure = failureCount.get();
        long avgTime = success > 0 ? totalFetchTime.get() / success : 0;
        
        System.out.printf("总数据源: %d%n", total);
        System.out.printf("成功抓取: %d (%.1f%%)%n", success, (double) success / total * 100);
        System.out.printf("失败抓取: %d (%.1f%%)%n", failure, (double) failure / total * 100);
        System.out.printf("平均抓取时间: %dms%n", avgTime);
        System.out.printf("总耗时: %dms%n", totalTime);
        System.out.printf("并发效率提升: %.1fx%n", calculateConcurrencyImprovement(totalTime, avgTime));
        
        System.out.println("\n📋 详细结果:");
        fetchResults.values().stream()
            .sorted(Comparator.comparing(FetchResult::getFetchTime).reversed())
            .forEach(result -> System.out.println("  " + result));
    }
    
    /**
     * 计算并发效率提升
     */
    private double calculateConcurrencyImprovement(long totalTime, long avgTime) {
        if (avgTime == 0) return 1.0;
        // 估算串行执行时间 = 平均时间 * 数据源数量
        long estimatedSerialTime = avgTime * DATA_SOURCES.size();
        return (double) estimatedSerialTime / totalTime;
    }
    
    /**
     * 关闭资源
     */
    public void shutdown() {
        System.out.println("\n🔄 关闭数据抓取系统...");
        executor.shutdown();
        try {
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        System.out.println("数据抓取系统已关闭");
    }
    
    /**
     * 获取实时统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalSources", DATA_SOURCES.size());
        stats.put("successCount", successCount.get());
        stats.put("failureCount", failureCount.get());
        stats.put("activeFetches", MAX_CONCURRENT_FETCHES - semaphore.availablePermits());
        stats.put("availablePermits", semaphore.availablePermits());
        stats.put("queueSize", ((ThreadPoolExecutor) executor).getQueue().size());
        return stats;
    }
    
    /**
     * 主方法
     */
    public static void main(String[] args) {
        ConcurrentDataFetchDemo demo = new ConcurrentDataFetchDemo();
        
        try {
            // 执行并发数据抓取
            Map<String, FetchResult> results = demo.fetchDataConcurrently();
            
            // 展示聚合结果
            System.out.println("\n📦 数据聚合结果");
            System.out.println("=".repeat(60));
            System.out.printf("成功抓取 %d 个数据源的数据%n", results.size());
            
            // 模拟数据聚合处理
            aggregateData(results);
            
        } finally {
            demo.shutdown();
        }
    }
    
    /**
     * 数据聚合处理
     */
    private static void aggregateData(Map<String, FetchResult> results) {
        System.out.println("\n🔍 开始数据聚合处理...");
        
        // 按数据类型分类
        Map<String, List<FetchResult>> categorizedData = new HashMap<>();
        
        for (FetchResult result : results.values()) {
            if (result.isSuccess()) {
                String category = determineCategory(result.getDataSource());
                categorizedData.computeIfAbsent(category, k -> new ArrayList<>()).add(result);
            }
        }
        
        // 输出分类结果
        System.out.println("数据分类结果:");
        categorizedData.forEach((category, dataList) -> {
            System.out.printf("  %s: %d 条数据%n", category, dataList.size());
        });
        
        // 生成聚合报告
        generateAggregationReport(categorizedData);
    }
    
    /**
     * 确定数据类别
     */
    private static String determineCategory(String dataSource) {
        if (dataSource.contains("用户")) return "用户数据";
        if (dataSource.contains("订单")) return "订单数据";
        if (dataSource.contains("商品")) return "商品数据";
        if (dataSource.contains("库存")) return "库存数据";
        if (dataSource.contains("支付")) return "支付数据";
        if (dataSource.contains("物流")) return "物流数据";
        return "其他数据";
    }
    
    /**
     * 生成聚合报告
     */
    private static void generateAggregationReport(Map<String, List<FetchResult>> categorizedData) {
        System.out.println("\n📈 数据聚合报告");
        System.out.println("=".repeat(40));
        
        int totalRecords = categorizedData.values().stream().mapToInt(List::size).sum();
        System.out.printf("总记录数: %d%n", totalRecords);
        
        categorizedData.forEach((category, dataList) -> {
            double avgFetchTime = dataList.stream()
                .mapToLong(FetchResult::getFetchTime)
                .average()
                .orElse(0.0);
            
            System.out.printf("%s: %d 条, 平均抓取时间: %.1fms%n", 
                category, dataList.size(), avgFetchTime);
        });
        
        System.out.println("\n✅ 数据聚合完成！");
    }
}